// Larbin
// Sebastien Ailleret
// 09-11-99 -> 09-03-02
#include <unistd.h>
#include <signal.h>
#include <time.h>
#include <netdb.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

#include "options.h"

#include "global.h"
#include "utils/text.h"
#include "fetch/checker.h"
#include "fetch/sequencer.h"
#include "fetch/fetchOpen.h"
#include "fetch/fetchPipe.h"
#include "interf/input.h"
#include "interf/output.h"
#include "utils/mypthread.h"

#include "utils/debug.h"
#include "utils/webserver.h"
#include "utils/histogram.h"
#include "utils/url.h"
// for gearman
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <libgearman/gearman.h>
//by me
#include <iostream>
#include <string>
using namespace std;

static void cron(gearman_job_st* job);

#ifdef PROF
static bool stop = false;
static void handler (int i) {
  stop = true;
}
#endif // PROF

// wait to limit bandwidth usage
#ifdef MAXBANDWIDTH
static void waitBandwidth (time_t *old,gearman_job_st* job) {
  while (global::remainBand < 0) {
    poll(NULL, 0, 10);
    global::now = time(NULL);
    if (*old != global::now) {
      *old = global::now;
      cron(job);
    }
  }
}
#else
#define waitBandwidth(x) ((void) 0)
#endif // MAXBANDWIDTH

#ifndef NDEBUG
static uint sta_count = 0;
#endif // NDEBUG

///////////////////////////////////////////////////////////
// If this thread terminates, the whole program exits
static void *larbin_main(gearman_job_st *job, void *context, size_t *result_size,
                         gearman_return_t *ret_ptr)
{
    cout<<"Received job:"<<gearman_job_handle(job)<<endl;
    string urlstr( (char*)gearman_job_workload(job) );
//    cout<<"workload:"<<start_url<<endl;
    cout<<"workload size:"<<gearman_job_workload_size(job)<<endl;    
    cout<<endl;
// create all the structures
    global glob(urlstr);
    
#ifdef PROF
    signal(2, handler);
#endif // PROF
#ifndef NOWEBSERVER
    // launch the webserver if needeed
    if (global::httpPort != 0)
        startThread(startWebserver, NULL);
#endif // NOWEBSERVER

    // Start the search
    printf("%s is starting its search\n", global::userAgent);
    time_t old = global::now;

    for (;;) {
        // update time
        global::now = time(NULL);
        if (old != global::now) {
            // this block is called every second
            old = global::now;
            cron(job);
        }
        stateMain(-sta_count);
        waitBandwidth(&old,job);
        stateMain(1);
        for (int i=0; i<global::maxFds; i++)
            global::ansPoll[i] = 0;
        for (uint i=0; i<global::posPoll; i++)
            global::ansPoll[global::pollfds[i].fd] = global::pollfds[i].revents;
        global::posPoll = 0;
        stateMain(2);
        input();                    // 从socket中读取url
        stateMain(3);
        sequencer();
        stateMain(4);
        fetchDns();
        stateMain(5);
        fetchOpen();
        stateMain(6);
        checkAll();
        // select
        stateMain(sta_count++);
        poll(global::pollfds, global::posPoll, 10);
        stateMain(7);
    }
    
    char* result=NULL;
	*result_size= strlen(result);
	*ret_ptr= GEARMAN_SUCCESS;
	return result;
}

static void usage(char *name)
{
  printf("\nusage: %s [-h <host>] [-p <port>]\n", name);
  printf("\t-h <host> - job server host\n");
  printf("\t-p <port> - job server port\n");
}

int main(int argc, char* argv[])
{
    string host("10.61.0.145");
    in_port_t port= 0;
    gearman_return_t ret;
    gearman_worker_st worker;
    int c;

    while ((c = getopt(argc, argv, "c:h:p:")) != -1)
    {
        switch(c)
        {
        case 'h':
            host = optarg;
            break;
        case 'p':
            port= (in_port_t)atoi(optarg);
            break;
        default:
            usage(argv[0]);
            exit(1);
        }
    }

    if (gearman_worker_create(&worker) == NULL)
    {
        fprintf(stderr, "Memory allocation failure on worker creation\n");
        exit(1);
    }
    cout<<"Worker created."<<endl;
    
    ret= gearman_worker_add_server(&worker, host.c_str(), port);
    if (ret != GEARMAN_SUCCESS)
    {
        fprintf(stderr, "%s\n", gearman_worker_error(&worker));
        exit(1);
    }
    cout<<"Worker added server:"<<host<<endl;

    ret= gearman_worker_add_function(&worker, "larbin", 0, larbin_main, NULL);
    if (ret != GEARMAN_SUCCESS)
    {
        fprintf(stderr, "%s\n", gearman_worker_error(&worker));
        exit(1);
    }
    cout<<"Worker added function of larbin."<<endl;

    while (1)
    {
        cout<<"Worker is ready..."<<endl;
        ret= gearman_worker_work(&worker);
        if (ret != GEARMAN_SUCCESS)
        {
            fprintf(stderr, "%s\n", gearman_worker_error(&worker));
            break;
        }else
        {
            break;
        }
    }

    gearman_worker_free(&worker);
    cout<<"Worker has been freed."<<endl;
}

// a lot of stats and profiling things
static void cron (gearman_job_st* job ) {
  // shall we stop
#ifdef PROF
  if (stop) exit(2);
#endif // PROF
  if (global::URLsDisk->getLength() == 0
      && global::URLsDiskWait->getLength() == 0
      && debUrl == 0)           // ?正处理的url
  {    
#ifdef EXIT_AT_END
      exit(0);
#endif // EXIT_AT_END
      gearman_return_t ret = gearman_job_send_complete(job,"Done",5);
      if (ret != GEARMAN_SUCCESS)
      {
          //fprintf(stderr, "%s\n", gearman_worker_error(&worker));
          cout<<"Error before complete."<<endl;
      }
  }
  

  // look for timeouts
  checkTimeout();
  // see if we should read again urls in fifowait
  if ((global::now % 300) == 0) {
    global::readPriorityWait = global::URLsPriorityWait->getLength();
    global::readWait = global::URLsDiskWait->getLength();
  }
  if ((global::now % 300) == 150) {
    global::readPriorityWait = 0;
    global::readWait = 0;
   }

#ifdef MAXBANDWIDTH
  // give bandwidth
  if (global::remainBand > 0) {
    global::remainBand = MAXBANDWIDTH;
  } else {
    global::remainBand = global::remainBand + MAXBANDWIDTH;
  }
#endif // MAXBANDWIDTH

#ifndef NOSTATS
  histoHit(pages, answers[success]);

  if ((global::now & 7) == 0) {
    urlsRate = (urls - urlsPrev) >> 3; urlsPrev = urls;
    pagesRate = (pages - pagesPrev) >> 3; pagesPrev = pages;
    successRate = (answers[success] - successPrev) >> 3;
    successPrev = answers[success];
    siteSeenRate = (siteSeen - siteSeenPrev) >> 3; siteSeenPrev = siteSeen;
    siteDNSRate = (siteDNS - siteDNSPrev) >> 3; siteDNSPrev = siteDNS;
#ifndef NDEBUG
    readRate = (byte_read - readPrev) >> 3; readPrev = byte_read;
    writeRate = (byte_write - writePrev) >> 3; writePrev = byte_write;
#endif // NDEBUG

#ifdef STATS
    printf("\n%surls : %d  (rate : %d)\npages : %d  (rate : %d)\nsuccess : %d  (rate : %d)\n",
           ctime(&global::now), urls, urlsRate, pages, pagesRate,
           answers[success], successRate);
#endif // STATS
  }
#endif // NOSTATS
}
